/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.caching.util;

import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;

import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;

/**
 * redis扩展工具
 *
 * @since 2020/2/21 23:35
 */
public abstract class RedisHelper {
    /**
     * scan 实现
     *
     * @param redisTemplate redisTemplate
     * @param pattern       表达式
     */
    public synchronized static Set<String> scan(RedisTemplate<String, Object> redisTemplate, String pattern) {
        RedisConnection redisConnection = redisTemplate.getConnectionFactory().getConnection();
        Set<String> resultSet = new HashSet<>();

        try {
            if (redisConnection instanceof RedisClusterConnection) {
                for (RedisClusterNode redisClusterNode : ((RedisClusterConnection) redisConnection).clusterGetNodes()) {
                    if (redisClusterNode.isMaster()) {
                        Cursor<byte[]> cursor = ((RedisClusterConnection) redisConnection).scan(redisClusterNode, new ScanOptions.ScanOptionsBuilder()
                                .match(pattern)
                                .count(10000)
                                .build());

                        while (cursor.hasNext()) {
                            resultSet.add(new String(cursor.next(), StandardCharsets.UTF_8));
                        }
                    }
                }

                return resultSet;
            } else {
                return redisTemplate.execute((RedisCallback<Set<String>>) connection -> {
                    Set<String> keysTmp = new HashSet<>();
                    try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder()
                            .match(pattern)
                            .count(10000).build())) {

                        while (cursor.hasNext()) {
                            keysTmp.add(new String(cursor.next(), StandardCharsets.UTF_8));
                        }
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    return keysTmp;
                });
            }

        } finally {
            redisConnection.close();
        }

    }
}

